﻿using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace KafkaPush.Sample
{
    class Program
    {
        const string _brokeList_dev = "127.0.0.1:9092";
        const string _brokeList_prd = "*:9092";
        const string TAG_PRD = "prd";
        const string TAG_DEV = "dev";
        const string TAG_PRODUCER = "p";
        const string TAG_CONSUMER = "c";

        static void Main(string[] args)
        {
            CancellationTokenSource cts = new CancellationTokenSource();
            Console.CancelKeyPress += (_, e) =>
            {
                e.Cancel = true; // prevent the process from terminating.
                cts.Cancel();
            };
            var enviromentTag = WaitKeyEnter("choose enviroment,prd or dev.", new List<string> { TAG_PRD, TAG_DEV });
            var brokes = enviromentTag == TAG_PRD ? _brokeList_prd : _brokeList_dev;
            Console.WriteLine($"connect to brokes:{brokes}");

            var topicName = WaitKeyEnter("enter your topic name", null);

            var producerOrConsumer = WaitKeyEnter("choose producer or consumer,enter p for producer.enter c for consumer.", new List<string> { TAG_PRODUCER, TAG_CONSUMER });

            if (string.Equals(producerOrConsumer, TAG_CONSUMER, StringComparison.CurrentCultureIgnoreCase))
                ConsumerSample(cts.Token, enviromentTag, topicName);
            else if (string.Equals(producerOrConsumer, TAG_PRODUCER, StringComparison.CurrentCultureIgnoreCase))
                ProduceSample(cts.Token, enviromentTag, topicName);
            else
                Console.WriteLine("error");

            Console.ReadLine();
        }

        static string WaitKeyEnter(string description, List<string> enterLimit)
        {
            Console.WriteLine(description);
            Console.Write(">");

            string text = Console.ReadLine();

            if (enterLimit == null) return text;

            while (!enterLimit.Any(x => string.Equals(x, text, StringComparison.CurrentCultureIgnoreCase)))
            {
                Console.WriteLine($"pls enter:{string.Join(" or ", enterLimit)}");
                Console.Write(">");
                text = Console.ReadLine();
            }

            return text;
        }


        static void ProduceSample(CancellationToken cancellationToken, string enviromentTag, string topicName)
        {
            var config = GetProduceConfigByEnviroment(enviromentTag);
            var builder = new ProducerBuilder<string, string>(config);
            using (var producer = builder.Build())
            {
                Console.WriteLine("\n-----------------------------------------------------------------------");
                Console.WriteLine($"Producer {producer.Name} producing on topic {topicName}.");
                Console.WriteLine("-----------------------------------------------------------------------");
                Console.WriteLine("To create a kafka message with UTF-8 encoded key and value:");
                Console.WriteLine("> key value<Enter>");
                Console.WriteLine("To create a kafka message with a null key and UTF-8 encoded value:");
                Console.WriteLine("> value<enter>");
                Console.WriteLine("Ctrl-C to quit.\n");

                while (!cancellationToken.IsCancellationRequested)
                {
                    Console.Write("> ");
                    string text;
                    try
                    {
                        text = Console.ReadLine();
                    }
                    catch (IOException)
                    {
                        // IO exception is thrown when ConsoleCancelEventArgs.Cancel == true.
                        break;
                    }
                    if (text == null)
                    {
                        // Console returned null before 
                        // the CancelKeyPress was treated
                        break;
                    }
                    string key = null;
                    string val = text;
                    // split line if both key and value specified.
                    int index = text.IndexOf(" ");
                    if (index != -1)
                    {
                        key = text.Substring(0, index);
                        val = text.Substring(index + 1);
                    }

                    var message = new Message<string, string> { Key = key, Value = val };

                    try
                    {
                        var resultTask = producer.ProduceAsync(topicName, message)
                            .ContinueWith(result =>
                            {
                                try
                                {
                                    var deliveryReport = result.Result;
                                    Console.WriteLine($"push {message.Value},partition:{deliveryReport.TopicPartition},offset:{deliveryReport.TopicPartitionOffset},status:{deliveryReport.Status}");
                                }
                                catch (Exception ex)
                                {
                                    Console.WriteLine($"push {message.Value},error:{ex.Message + ex.StackTrace}");
                                }
                            });

                        resultTask.Wait();
                    }
                    catch (ProduceException<string, string> e)
                    {
                        Console.WriteLine($"failed to deliver message: {e.Message} [{e.Error.Code}]");
                    }
                }
            }
        }
        /*
         *  自动 OR 手动
         *  前面所说的自动分配是指在 KafkaConsumer API中的subscribe()方法。这个方法强制要求你为消费者设置一个消费者组，group.id参数不能为空。
         *  而你不需要处理分区的分配问题。
         *  而对应subscribe()方法，你可以采用手动的方式，指定消费者读取哪个主题分区，则：assign() 方法。
         *  当你需要精确地控制消息处理的负载，也能确定哪个分区有哪些消息时，这种手动的方式会很有用。
         *  但这时Kafka也无法提供rebalancing的功能了。而且在使用手动的方式时，你可以不指定消费者组，group.id为空。
         *  两种方式都各有适用场景，但同时不建议同时使用两种方式，这会带来风险。假设一个消费者组G1，组内只有一个消费者C1，
         *  订阅subscribe了一个具有两个分区P1、P2的主题T1。这时在G1新增一个消费者C2，用assign的方式关联P1和P2。视乎一切都可行，但其实是糟糕的情况。
         *  本质上，使用场景被混淆了，你无法确定G1是在共同协助还是在进行发布/订阅。实际使用中，offset的提交格式是这样的：
         * 
         * */
        static void ConsumerSample(CancellationToken cancellationToken, string enviromentTag, string topicName)
        {
            var config = GetConsumerConfigByEnviroment(enviromentTag);

            using (var consumer = new ConsumerBuilder<string, string>(config)
                .SetErrorHandler((_, e) => Console.WriteLine($"error:{e.Reason}"))
                .SetStatisticsHandler((_, json) => Console.WriteLine($"statistics:{json}"))
                .SetPartitionsAssignedHandler((c, partitions) => Console.WriteLine($"assigned partitions:[{string.Join(",", partitions)}]"))
                .SetPartitionsRevokedHandler((c, partitions) => Console.WriteLine($"revoking assignment:[{string.Join(",", partitions)}]"))
                .Build())
            {
                consumer.Subscribe(topicName);
                try
                {
                    //int i = 0;
                    while (true)
                    {
                        var consumeResult = consumer.Consume(cancellationToken);
                        if (consumeResult.IsPartitionEOF)
                        {
                            Console.WriteLine($"Reached end of topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
                            continue;
                        }
                        Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Message.Value}");
                        //i++;
                        //if (i == 10)
                        //{
                        //    //中途可以再次subs
                        //    consumer.Subscribe("myTopic");
                        //}
                    }
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"consumer error:{ex.Message + ex.StackTrace}");
                }
            }
        }

        static ProducerConfig GetProduceConfigByEnviroment(string enviromentTag)
        {
            var config = new ProducerConfig
            {
                RequestTimeoutMs = 3000,
                MessageTimeoutMs = 10000,
                //SecurityProtocol = SecurityProtocol.SaslPlaintext,
                //SaslMechanism = SaslMechanism.Plain
            };
            if (enviromentTag == TAG_DEV)
            {
                config.BootstrapServers = _brokeList_dev;
                //config.SaslUsername = "kafka";
                //config.SaslPassword = "kafka#secret";
            }
            else if (enviromentTag == TAG_PRD)
            {
                config.BootstrapServers = _brokeList_prd;
                config.SaslUsername = "passport";
                config.SaslPassword = "!0ka$!uhukFCDB88";
            }
            else
            {
                Console.WriteLine($"error enviromentTag:{enviromentTag}");
            }

            return config;
        }

        static ConsumerConfig GetConsumerConfigByEnviroment(string enviromentTag)
        {
            var config = new ConsumerConfig
            {
                EnableAutoCommit = true,
                GroupId = "sample-consumer",
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnablePartitionEof = true,
                SecurityProtocol = SecurityProtocol.SaslPlaintext,
                SaslMechanism = SaslMechanism.Plain,
            };
            if (enviromentTag == TAG_DEV)
            {
                config.BootstrapServers = _brokeList_dev;
                config.SaslUsername = "kafka";
                config.SaslPassword = "kafka#secret";
            }
            else if (enviromentTag == TAG_PRD)
            {
                config.BootstrapServers = _brokeList_prd;
                config.SaslUsername = "passport";
                config.SaslPassword = "!0ka$!uhukFCDB88";
            }
            else
            {
                Console.WriteLine($"error enviromentTag:{enviromentTag}");
            }

            return config;
        }
    }
}
